package com.example.batch.task;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.batch.core.ExitStatus;
import org.springframework.batch.core.StepContribution;
import org.springframework.batch.core.StepExecution;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.ExecutionContext;
import org.springframework.batch.item.ReaderNotOpenException;
import org.springframework.batch.item.file.DefaultBufferedReaderFactory;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.NonTransientFlatFileException;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.core.io.PathResource;
import org.springframework.util.CollectionUtils;

import java.io.BufferedReader;
import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;


/**
 * @author LiYuhang
 * @version 0.1
 * @application
 * @Date 2021/7/17 16:52
 * @Description TODO
 */
@Slf4j
public class LineReaderTasklet implements Tasklet, StepExecutionListener {

    private long startTime;

    private BufferedReader reader;
    private String resource;
    private List<String> batch;

    private final int BATCH_LIMIT = 10;

    private Long currLine = -1L;
    private Long offerCount = -1L;

    private ExecutionContext executionContext;

    private ArrayBlockingQueue<List<String>> blockingQueue;

    public LineReaderTasklet(String filePath,ArrayBlockingQueue blockingQueue) {
        this.resource = filePath;
        this.blockingQueue = blockingQueue;
        try {
            PathResource resource = new PathResource(filePath);
            reader = new DefaultBufferedReaderFactory().create(resource,FlatFileItemReader.DEFAULT_CHARSET);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void beforeStep(StepExecution stepExecution) {
        log.info("start");
        executionContext = stepExecution.getJobExecution().getExecutionContext();
        currLine = executionContext.getLong("currLine",0);
        startTime = System.currentTimeMillis();
    }

    @Override
    public RepeatStatus execute(StepContribution contribution,ChunkContext chunkContext) throws Exception {
        batch = new LinkedList<>();

        if (currLine > 0) {
            log.warn("currLine :{} skipt",currLine);
            reader.skip(currLine);
        }

        while (true) {
            String cs = readLine();
            if (StringUtils.isBlank(cs)) {
                break;
            }
            batch.add(cs.trim());
            currLine++;
            ensureToSend();
        }
        log.info(" contribution:{}  chunkContext:{}",contribution,chunkContext);
        return RepeatStatus.FINISHED;
    }

    private void ensureToSend() {
        if (batch.size() >= BATCH_LIMIT) {
            blockingQueue.offer(batch);
            executionContext.putLong("currLine",currLine);
            batch = new LinkedList<>();
            offerCount++;
        }
    }


    @Override
    public ExitStatus afterStep(StepExecution stepExecution) {
        if (!CollectionUtils.isEmpty(batch)) {
            blockingQueue.offer(batch);
        }
        executionContext.putLong("currLine",currLine);
        log.warn(" currentLine size:{} offerCount:{} time consuming :{}",currLine,offerCount,System.currentTimeMillis() - startTime);

        return ExitStatus.COMPLETED;
    }

    private String readLine() {

        if (reader == null) {
            throw new ReaderNotOpenException("Reader must be open before it can be read.");
        }

        String line = null;

        try {
            line = this.reader.readLine();
            if (line == null) {
                return null;
            }
        } catch (IOException e) {
            // Prevent IOException from recurring indefinitely
            // if client keeps catching and re-calling
            throw new NonTransientFlatFileException("Unable to read from resource: [" + resource + "]",line);
        }
        return line;
    }
}
